package org.foreveross.hbase.client;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.log4j.Logger;

import com.fare.taluo.vo.core.HBaseFilterVO;

public class HBaseHelper {
	
	private static Logger log = Logger.getLogger(HBaseHelper.class);
	/**
	 * 通过一个rowkey取得对应的记录
	 * 
	 * @param tableName
	 *            ：表名
	 * @param rowKey
	 * @return
	 * @throws IOException
	 */
	public static Result getOneRecord(String tableName, String rowKey)
			throws IOException {
		Get get = null;
		Result rs = null;
		HTableInterface table = null;
		try{
			get = new Get(rowKey.getBytes("utf-8"));
		    table = HbaseDriverManager.getHtable(tableName);
			rs = table.get(get);
		}finally{
			HbaseDriverManager.releaseHtable(table);
			 get = null;
			 rs = null;
			 table = null;
		}
		return rs;
	}
	/**
	 * 将hbase中的一行记录转换为方便统计的Map结构
	 * 
	 * @param res
	 *            Map<列簇，Map<列名，列值>>
	 * @return
	 */
	public static Map<String, Map<String, String>> transformHBaseRowToMap(
			Result res) {
		Map<String, Map<String, String>> temp = new HashMap<String, Map<String, String>>();// 保存HBASE中存储的信息
		for (KeyValue kv : res.raw()) {
		
			if (temp.get(new String(kv.getFamily())) != null) {
				temp.get(new String(kv.getFamily())).put(
						new String(kv.getQualifier()),
						new String(kv.getValue()));
			} else {
				Map<String, String> colVal = new HashMap<String, String>();
				colVal.put(new String(kv.getQualifier()),
						new String(kv.getValue()));
				temp.put(new String(kv.getFamily()), colVal);
			}
		}
		return temp;
	}

	/**
	 * 根据条件查询hbase数据库
	 * 
	 * @param tableName
	 *            :表名
	 * @param startRowKey
	 *            ：开始rowkey
	 * @param stopRowKey
	 *            ：结束rowkey
	 * @param family
	 *            ：列簇
	 * @return
	 */
	public static ResultScanner getRecords(String tableName,
			String startRowKey, String stopRowKey, String[] family)
			throws IOException {
		return HBaseHelper.getRecords(tableName, startRowKey, stopRowKey,
				family, null);
	}

	/**
	 * 根据条件查询hbase数据库
	 * 
	 * @param tableName
	 *            :表名
	 * @param startRowKey
	 *            ：开始rowkey
	 * @param stopRowKey
	 *            ：结束rowkey
	 * @param filters
	 *            ：过滤器
	 * @return
	 */
	public static ResultScanner getRecords(String tableName,
			String startRowKey, String stopRowKey) throws IOException {
		return HBaseHelper.getRecords(tableName, startRowKey, stopRowKey, null, null);
	}

	/**
	 * 根据条件查询hbase数据库
	 * 
	 * @param tableName
	 *            :表名
	 * @param startRowKey
	 *            ：开始rowkey
	 * @param stopRowKey
	 *            ：结束rowkey
	 * @param family
	 *            ：列簇
	 * @param filters
	 *            ：过滤器
	 * @return
	 */
	public static ResultScanner getRecords(String tableName,
			String startRowKey, String stopRowKey, String[] family,
			List<Filter> filters) throws IOException {
		long start = System.currentTimeMillis();
		ResultScanner ss = null;
		Scan s = null;
		HTableInterface table = null;
		try{
			 s = new Scan();
			s.setCaching(500);
			 table = HbaseDriverManager.getHtable(tableName);
			if (StringUtils.isNotBlank(startRowKey)) {
				s.setStartRow(startRowKey.getBytes());
			}
			if (StringUtils.isNotBlank(stopRowKey)) {
				s.setStopRow(stopRowKey.getBytes());
			}
			if (StringUtils.isNotBlank(stopRowKey)) {
				s.setStopRow(stopRowKey.getBytes());
			}
			if (family != null && family.length > 0) {
				for (String cf : family) {
					s.addFamily(cf.getBytes());
				}
			}
			if (filters != null && filters.size() > 0) {
				FilterList list = new FilterList(filters);
				s.setFilter(list);
			}
			ss = table.getScanner(s)		;
		}finally{
			HbaseDriverManager.releaseHtable(table);
			 ss = null;
			 s = null;
			 table = null;
		}
		log.debug("tableName:"+tableName  +"  startRowKey: "+startRowKey  +"   stopRowKey:"+stopRowKey+"    共用时(ms)："+(System.currentTimeMillis()-start));
		return ss;
	}


	/**
	 * 通过rowkey取得对应的记录
	 * 
	 * @param tableName
	 *            ：表名
	 * @param rowKey
	 * @return
	 * @throws IOException
	 */
	public static Result[] getRecords(String tableName, List<String> rowKeys)
			throws IOException {
		List<Get> list = new ArrayList<Get>();
		for(String s : rowKeys){
			list.add( new Get(s.getBytes("utf-8")));
		}
		
		Result[] rs = null;
		HTableInterface table = HbaseDriverManager.getHtable(tableName);
		rs = table.get(list);
		
		table.close();

		return rs;
	}
	public static void main(String[] agrs) throws Exception {
		getOneRecord("CRAWL_DETAIL_INFO","000010012014101400499558");
				 
	}


	public static ResultScanner getRecordsTest(String tableName,
			String startRowKey, String stopRowKey, 
			List<HBaseFilterVO> voList,List<Filter> filters) throws IOException {
		long start = System.currentTimeMillis();
		
		ResultScanner ss = null;
		Scan s = null;
		HTableInterface table = null;
		try{
			s = new Scan();
			s.setCaching(500);
			table = HbaseDriverManager.getHtable(tableName);
			if (StringUtils.isNotBlank(startRowKey)) {
				s.setStartRow(startRowKey.getBytes());
			}
			if (StringUtils.isNotBlank(stopRowKey)) {
				s.setStopRow(stopRowKey.getBytes());
			}
			if (StringUtils.isNotBlank(stopRowKey)) {
				s.setStopRow(stopRowKey.getBytes());
			}
			if(com.fare.taluo.util.StringUtils.checkList(voList)){
				for (HBaseFilterVO vo : voList) {
			      	  //没有指定列，则默认指定列簇
			      	  if(null == vo.getQualifier() && null != vo.getFamily()){
			      		  s.addFamily(vo.getFamily().getBytes());
			      	  }else if(null != vo.getQualifier()){
			      		  s.addColumn(vo.getFamily().getBytes(), vo.getQualifier().getBytes());
			      	  }
			      	  
			      	  
			        }
			}
	
			if (filters != null && filters.size() > 0) {
				FilterList list = new FilterList(filters);
				s.setFilter(list);
			}
			ss = table.getScanner(s);
		}finally{
			HbaseDriverManager.releaseHtable(table);
			 ss = null;
			 s = null;
			 table = null;
		}
		
		log.debug("tableName:"+tableName  +"  startRowKey: "+startRowKey  +"   stopRowKey:"+stopRowKey+"    共用时(ms)："+(System.currentTimeMillis()-start));
		return ss;
	}

}
